package org.zjvis.datascience.spark.util;

import org.apache.commons.lang3.StringUtils;
import org.apache.spark.sql.AnalysisException;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.catalog.Catalog;
import org.apache.spark.sql.catalog.Column;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @description Hive数据读取工具类
 * @date 2021-12-23
 */
public class HiveTool {

    private final static Logger logger = LoggerFactory.getLogger("HiveTool");

    public final static String DATA_SCHEMA = "dataset";

    public boolean createDatabase(SparkSession sparkSession, String database) {
        if (StringUtils.isEmpty(database) || sparkSession == null) {
            logger.error("param is empty!!!");
            return false;
        }
        String sql = "CREATE DATABASE IF NOT EXISTS %s";
        try {
            sparkSession.sql(String.format(sql, database));
        } catch (Exception exception) {
            logger.error(exception.getMessage());
            return false;
        }
        return true;
    }

    public static Map<String, String> getTableMetaMap(SparkSession sparkSession, String tableName) {
        Map<String, String> meta = new HashMap<>();
        Catalog catalog = sparkSession.catalog();
        String schema;
        String table = tableName;
        if (!tableName.contains(".")) {
            schema = DATA_SCHEMA;
        } else {
            String[] tmps = tableName.split("\\.");
            schema = tmps[0];
            table = tmps[tmps.length - 1];
        }
        List<Column> columns;
        try {
            columns = catalog.listColumns(schema, table).collectAsList();
            for (Column column : columns) {
                meta.put(column.name(), column.dataType());
            }
        } catch (AnalysisException e) {
            return meta;
        }
        return meta;
    }

    public boolean createTable(SparkSession sparkSession, String tableName) {
        return true;
    }

}
